package com.simple.datax.conn.impl;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.PrintWriter;
import java.security.KeyStore;

import javax.net.ssl.KeyManager;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;

import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPFile;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.commons.net.ftp.FTPSClient;
import org.apache.commons.net.util.TrustManagerUtils;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.SimpleMessage;
import com.simple.datax.api.ConnectorException;
import com.simple.datax.conn.BaseConnector;

/** 
 * @Title: FTPConnImpl.java 
 * @Package com.simple.datax.conn.impl 
 * @Description: 实现具体的FTP/FTPS连接操作，包括连接，接收，发送等。 
 * @author Jin Yi 
 * @date 2012-12-30
 * @version V1.0 
 */
public class FTPConnImpl extends BaseConnector {
	private static final Logger logger = LoggerFactory
			.getLogger(FTPConnImpl.class);
	public static final String TYPE = "FTP";
	//	private static final int FTP_LIST_PAGE_SIZE = 10;
	private FTPClient ftpClient = null;
	private String host;
	private int port=21;
	private String username;
	private String password;
	private String workingPath = null;

	private boolean binaryTransfer = true;
	private boolean passiveMode = true;
//	private String encoding = "UTF-8";
//	private int clientTimeout = 1000 * 30;
//	private int dataTimeout = 1000 * 30;
	private String protocol = "NO"; // default is Explicit Mode
	private String trustmgr = null;
	private String keymgr = null;
	private boolean bSSL = false;
	private String trust_path = null;
	private String key_path = null;
	private String key_pw = null;
	private String trust_pw = null;
	private boolean needMove = false;
	private String destDir = null;
	private String homeDir = null;
	
	ByteArrayOutputStream commandLog = new ByteArrayOutputStream();
	PrintWriter commandPrinter = new PrintWriter(commandLog);
	
	private static final String KEYSTORE_DIR = "../keystore/";
	
	/**
	 * 构造器
	 * @param connPara
	 * @throws ConnectorException
	 */
	public FTPConnImpl(String connPara) throws ConnectorException {
		this(connPara, false);
	}
	
	/**
	 * 构造器
	 * @param connPara
	 * @param bSSL 是否使用SSL
	 * @throws ConnectorException
	 */
	public FTPConnImpl(String connPara, boolean bSSL) throws ConnectorException {
		super(connPara);
		this.bSSL = bSSL;
		setConnPara(connPara);
		parseConnPara();
	}
	
	/**
	 * 准备FTPClient对象，连接并登录FTP Server
	 * 
	 * @throws ConnectorException
	 */
	private void prepareFTPClient() throws ConnectorException {
		if (ftpClient != null && ftpClient.isConnected()) {
			
			//检测ftpClient是否还有效
			try {
				ftpClient.noop();
				return;
			} catch (Exception ex) {
				try {
					ftpClient.disconnect();
				} catch (Exception e) {
					
				}
				logger.debug("FTP Client needs re-init. ");
			}
		}
		createFTPClient(); // 构造一个FtpClient实例
		ftpClient.setControlEncoding(getEncoding()); // 设置字符集

		connect();

		// 设置为passive模式
		if (passiveMode) {
			ftpClient.enterLocalPassiveMode();
		}
		setFileType(); // 设置文件传输类型

//		try {
//			ftpClient.setSoTimeout(clientTimeout);
//			ftpClient.setDataTimeout(dataTimeout);
//			
//			
//		} catch (SocketException e) {
//			throw new ConnectorException("Set timeout error.", e);
//		}
		changeWorkingDirectory();
	}

	/**
	 * 创建FTPClient
	 * @throws ConnectorException 
	 */
	private void createFTPClient() throws ConnectorException {
		if (!bSSL) {
			ftpClient = new FTPClient();
		} else {
			FTPSClient ftps;
			if (protocol.equalsIgnoreCase("YES")) {
				ftps = new FTPSClient(true); // Implicit Mode
			} else if (protocol.equalsIgnoreCase("NO")) {
				ftps = new FTPSClient(false); // Explicit Mode
			} else {
				ftps = new FTPSClient(false); // default is Explicit Mode
			}
//			 else {
//				String prot[] = protocol.split(",");
//				if (prot.length == 1) { // Just protocol
//					ftps = new FTPSClient(protocol);
//				} else { // protocol,true|false
//					ftps = new FTPSClient(prot[0],
//							Boolean.parseBoolean(prot[1]));
//				}
//			}
			ftps.setDefaultTimeout(10000);
			ftps.addProtocolCommandListener(new org.apache.commons.net.PrintCommandListener(
					commandPrinter));
			if ("all".equals(trustmgr)) {
				ftps.setTrustManager(TrustManagerUtils
						.getAcceptAllTrustManager());
			} else if ("valid".equals(trustmgr)) {
				try {
					ftps.setTrustManager(getTrustManager());
				} catch (Exception e) {
					throw new ConnectorException("Set trust manager failed", e);
				}
			} else if ("none".equals(trustmgr)) {
				ftps.setTrustManager(null);
			}
			
			if ("valid".equals(keymgr)) {
				try {
					ftps.setKeyManager(getKeyManager());
				} catch (Exception e) {
					throw new ConnectorException("Set trust manager failed", e);
				}
			} else if ("none".equals(keymgr)) {
				ftps.setKeyManager(null);
			}
			
			ftpClient = ftps;
		}
	}
	
	
	/**
	 * 连接到ftp服务器
	 * 
	 * @return 连接成功返回true，否则返回false
	 * @throws ConnectorException
	 */
	private boolean connect() throws ConnectorException {
		try {
			ftpClient.connect(host, port);
			
			// 连接后检测返回码来校验连接是否成功
			int reply = ftpClient.getReplyCode();

			if (FTPReply.isPositiveCompletion(reply)) {
				// 登陆到ftp服务器
				if (ftpClient.login(username, password)) {
					homeDir = ftpClient.printWorkingDirectory();
					return true;
				}
			} else {
				ftpClient.disconnect();
				throw new ConnectorException("FTP server refused connection.");
			}
		} catch (Exception e) {
			if (ftpClient.isConnected()) {
				try {
					ftpClient.disconnect(); // 断开连接
				} catch (IOException e1) {
					logger.debug("Connector login failed and disconnect it failed.");
				}

			}
			throw new ConnectorException("Could not connect to server.", e);
		} finally {
			// 打印FTP连接时的具体连接日志，包括FTPS连接是的TLS认证过程
			logger.debug((commandLog.toString()));
		}
		return false;
	}
	
	/**
	 * 断开ftp连接
	 * 
	 * @throws ConnectorException
	 */
	@Override
	public void disconnect() {
		try {
			ftpClient.logout();
			if (ftpClient.isConnected()) {
				ftpClient.disconnect();
				ftpClient = null;
			}
		} catch (IOException e) {
			logger.warn("Could not disconnect from server.", e);
//			throw new ConnectorException("Could not disconnect from server.", e);
		}
	}

	/**
	 * 设置文件传输类型
	 * 
	 * @throws ConnectorException
	 * @throws IOException
	 */
	private void setFileType() throws ConnectorException {
		try {
			if (binaryTransfer) {
				ftpClient.setFileType(FTPClient.BINARY_FILE_TYPE);
			} else {
				ftpClient.setFileType(FTPClient.ASCII_FILE_TYPE);
			}
		} catch (IOException e) {
			throw new ConnectorException("Could not to set file type.", e);
		}
	}

	@Override
	public String getMessage() throws ConnectorException {
		String ret = null;
		try {
			prepareFTPClient();
			FTPFile[] files = null;
			try {
				files = ftpClient.listFiles();
			} catch (Exception e1) {
				throw new ConnectorException("Going through directory failed.", e1);
			}
			if (files != null && files.length>0) {
				for (FTPFile file:files) {
					if (file.isFile()) { // 处理文件（跳过目录等特殊文件）
						String name = file.getName();
						ByteArrayOutputStream baos = new ByteArrayOutputStream();
						try {
							ftpClient.retrieveFile(name, baos);
							getReplyCode();
							ret = baos.toString(getEncoding());
							
							// 如果文件正在上传过程中，则baos长度应该为0，此时，跳过此文件
							if (ret == null || ret.length()==0) {
								continue;
							}

						} catch (IOException e) {
							throw new ConnectorException("Get remote file failed", e);
						} finally {
							if (baos != null) {
								try {
									baos.close();
								} catch (IOException e) {
		
								}
							}
						}
		
						// 下载文件后，删除FTP服务器上的原始文件
						try {
							ftpClient.deleteFile(name);
							getReplyCode();
						} catch (IOException e) {
							throw new ConnectorException("Can't delete file - "
									+ name, e);
						}
						
						// 一次只取一个文件，所以跳出
						break;
					}
				}
			}
		} catch (Exception e) {
			throw new ConnectorException("Get remote file failed", e);
		}
		return ret;
	}
	
//	protected String findFileToProcess() throws Exception {
//		FTPListParseEngine engine = ftpClient.initiateListParsing();
//		FTPFile[] files = null;
//		while (engine.hasNext()) {
//			files = engine.getNext(FTP_LIST_PAGE_SIZE);
//			if (files == null) {
//				break;
//			}
//			// FilenameFilter filenameFilter = getFilenameFilter();
//			for (int i = 0; i < files.length; i++) {
//				FTPFile file = files[i];
//				if (file.isFile()) {
//					return file.getName();
//				}
//			}
//		}
//		getReplyCode();
//		return null;
//	}


	/**
	 * 解析连接参数
	 * @throws ConnectorException 
	 */
	private void parseConnPara() throws ConnectorException {
		//connPara = "IP:PORT:USER:PASSWORD:PATH";
		String[] tmpStr = splitConnPara(connPara);
		if (tmpStr.length==14) {
			host = tmpStr[0];
			try {
				port = Integer.parseInt(tmpStr[1]);
			} catch (NumberFormatException e) {
				logger.error("Connector port is invalid." + tmpStr[1], e);
			}
			username = tmpStr[2];
			password = tmpStr[3];
			workingPath = tmpStr[4];
			protocol = tmpStr[5];
			keymgr = tmpStr[6];
			trustmgr = tmpStr[7];
			key_path = tmpStr[8];
			trust_path = tmpStr[9];
			key_pw = tmpStr[10];
			trust_pw = tmpStr[11];
			if ("YES".equalsIgnoreCase(tmpStr[12])) {
				needMove = true;
			}
			destDir = tmpStr[13];
		} else if (tmpStr.length == 7) {
			host = tmpStr[0];
			port = Integer.parseInt(tmpStr[1]);
			username = tmpStr[2];
			password = tmpStr[3];
			workingPath = tmpStr[4];
			if ("YES".equalsIgnoreCase(tmpStr[5])) {
				needMove = true;
			}
			destDir = tmpStr[6];
		} else {
			throw new ConnectorException(
					"Parsing FTPClient parameters failed. Please check parameter - "
							+ connPara);
		}
	}

//	/**
//	 * 发送所有队列中的报文 
//	 * @throws ConnectorException 
//	 */
//	protected boolean send() throws ConnectorException {
//		SimpleMessage simple = null;
//		try {
//			prepareFTPClient();
//		} catch (ConnectorException e) {
//			throw new ConnectorException("FTPClient initialization failed.");
//		}
//		for (SimpleObject object:messageList) {
//			simple = ((SimpleMessage)object);
//			try {
//				put(String.valueOf(fileCountName++)+".xml", simple.getContent());
////				put(simple.getFileName(), simple.getContent());
////				Thread.sleep(2000);
//				logger.debug("FTP Send message - " + simple.getSentCommunType()
//						+ simple.getSentCommunParam());
//			} catch (Exception ex) {
//				logger.error("Fail to send out message.", ex);
//			}
//		}
////		messageList.removeAllElements();
//		return true;
//	}

	@Override
	public String getConnType() {
		if (bSSL) {
			return TYPE+"S"; 
		} else {
			return TYPE;
		}
	}
	
	/**
     * 上传一个本地文件到远程指定文件
     * 
     * @param remoteAbsoluteFile 远程文件名(包括完整路径)
     * @param content 文件内容
     * @return 成功时，返回true，失败返回false
     * @throws ConnectorException
	 * @throws IOException 
     */
    private void put(String remoteAbsoluteFile, String content) throws ConnectorException, IOException {
    	ByteArrayInputStream input = null;
    	boolean ret = false;
        if (remoteAbsoluteFile==null || remoteAbsoluteFile.isEmpty() || content==null) {
        	throw new ConnectorException("Remote file name is null or content is null.");
        }
        try {
            // 处理传输
        	input = new ByteArrayInputStream(content.getBytes(getEncoding())); 
        	ret = ftpClient.storeFile(remoteAbsoluteFile, input);
        	if (!ret) {
        		throw new ConnectorException("Uploading File Failed. Reply Code:"+ftpClient.getReplyCode()+". Please check error log for detail reply.");
        	}
            getReplyCode();
//            return true;
        } catch (FileNotFoundException e) {
            throw e;
        } catch (IOException e) {
            throw e;
        } finally {
            try {
                if (input != null) {
                    input.close();
                }
            } catch (Exception e) {
                logger.error("Close file input stream failed.", e);
            }
        }
    }
    
    private void move(String srcAbsoluteFile, String destAbsoluteFile) throws ConnectorException, IOException {
    	boolean ret = false;
		ret = ftpClient.rename(srcAbsoluteFile, destAbsoluteFile);
		getReplyCode();
		if (!ret) {
    		throw new ConnectorException("Moving File Failed. Reply Code:"+ftpClient.getReplyCode()+". Please check error log for detail reply.");
    	}
    }
    
    /**
     * 切换工作目录
     * @throws ConnectorException
     */
    private void changeWorkingDirectory() throws ConnectorException {
    	// Changes working directory
    	boolean success = false;
    	try {
    		if (workingPath == null || workingPath.trim().isEmpty()) {
    			// not change directory, return true
    			success = true;
    		} else {
    			success = ftpClient.changeWorkingDirectory(workingPath);
    		}
    	} catch (IOException ex) {
    		throw new ConnectorException("Changing working directory failed", ex);
    	}
        int reply = ftpClient.getReplyCode();
        if (!FTPReply.isPositiveCompletion(reply)) {
        	logger.debug("FTP reply is " + reply);
        }
        if (!success) {
        	throw new ConnectorException("Failed to change working directory. See server's reply.");
        }
    }
    
    /**
     * 检查FTP操作的返回值
     */
    private void getReplyCode() {
    	int reply = ftpClient.getReplyCode();
    	if (!FTPReply.isPositiveCompletion(reply)) {
    		logger.error("Reply Code: " + reply);
    	}
    }
    
    
    static int fileCountName = 2;

	@Override
	protected void prepareSend() throws ConnectorException {
		prepareFTPClient();
	}

	@Override
	protected void internalSend(SimpleMessage message) throws ConnectorException {
		try {
			String fileName = message.getOriFileName()+"."+message.getSentMsgFormat();
			put(fileName, message.getContent());
			if (needMove) {
				move(fileName, homeDir+File.separator+destDir+File.separator+fileName);
			}
			
		} catch (IOException ex) {
			throw new ConnectorException("FTP internalSend failed", ex, true);
		} catch (Exception ex) {
			throw new ConnectorException("FTP internalSend failed", ex);
		}
	}
	
	/**
	 * 获取 KeyManager 用于读取客户端证书，发给服务器做认证
	 * @return
	 * @throws Exception
	 */
	private KeyManager getKeyManager() throws Exception {
		String keyfile = KEYSTORE_DIR + key_path;
		KeyStore key_ks = KeyStore.getInstance("JKS");
		key_ks.load(new FileInputStream(keyfile), key_pw.toCharArray());

		KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory
				.getDefaultAlgorithm());
		kmf.init(key_ks, key_pw.toCharArray());
		KeyManager[] km = kmf.getKeyManagers();
		// TODO: 一个KeyStore文件如果包含多个key，需要根据alias取到相应的KeyManager
		return km[0];
	}

	/**
	 * 获取 TrustManager 用于验证服务器的证书是否可信任
	 * @return
	 * @throws Exception
	 */
	private TrustManager getTrustManager() throws Exception {
		String keyfile = KEYSTORE_DIR + trust_path;
		KeyStore trust_ks = KeyStore.getInstance("JKS");
		trust_ks.load(new FileInputStream(keyfile), trust_pw.toCharArray());

		TrustManagerFactory tf = TrustManagerFactory
				.getInstance(TrustManagerFactory.getDefaultAlgorithm());
		tf.init(trust_ks);
		
		TrustManager[] tm = tf.getTrustManagers();
		// TODO: 一个KeyStore文件如果包含多个certification，需要根据alias取到相应的TrustManager
		return tm[0];
	}

	@Override
	protected void prepareReceive() throws ConnectorException {
		prepareFTPClient();
	}
}
